import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileOutputStream;
import java.net.URI;

/**
 * @author qiss
 * @create 2018-02-12 15:39
 **/
public class HdfsClient {
    FileSystem fs = null;

    @Before
    public void init() throws Exception{
        /**
         * 构造一个配置参数对象，设置一个参数：我们要访问的hdfs的URI
         * 通过FileSystem.get()方法去构造一个访问hdfs的客户端，以及hdfs的访问地址
         * new Configuretion()的时候，他就会去加载jar中的hdfs-default.xml，然后再加载classpath下的hdfs-site.xml
         * 参数优先级：1.客户端代码设置的值，2.classpath下的用户自定义配置文件，3.然后是服务器的默认配置
         */
        Configuration conf = new Configuration();
        //conf.set("fs.defaultFS","hdfs://hadoop-namenode1:9000");
        //conf.set("dfs.replication","3");
        //fs = FileSystem.get(conf);
        //如果这样去获取，那conf里面就可以不设置“fs.defaultFS”参数，而且，这个客户端的身份标识已经是root用户
        fs = FileSystem.get(new URI("hdfs://hadoop-namenode1:9000"),conf,"root");
    }

    /**
     * 从本地向hdfs上传文件
     * @throws Exception
     */
    @Test
    public void testPutFileToHdfs() throws Exception{
        Path src = new Path("D:\\hdfsTest.txt");
        Path dst = new Path("/wordcount");
        fs.copyFromLocalFile(src,dst);
        fs.close();
    }

    /**
     * 从hdfs复制文件到本地，下载是需要自行shell命令，但是win环境下有问题
     * @throws Exception
     */
    @Test
    public void testDownloadFileToLocal() throws Exception{
        //fs.copyToLocalFile(new Path("/README.txt"),new Path("d:/"));//会报错
        fs.copyToLocalFile(false,new Path("/README.txt"),new Path("d:/"),true);
        fs.close();
    }

    public void testMkdirAndDeleteAndRename() throws Exception{
        //创建目录
        fs.mkdirs(new Path("/wordcount/textDate"));
        //删除文件夹，如果是非空文件夹，参数2必须是true
        fs.delete(new Path("/aaa"),true);
        //重命名文件或文件夹
        fs.rename(new Path("/a1"),new Path("/a2"));
    }

    @Test
    public void testListFiles()throws Exception{
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/wordcount"), true);
        while(listFiles.hasNext()){
            LocatedFileStatus fileStatus = listFiles.next();
            System.out.println(fileStatus.getPath().getName());
            System.out.println(fileStatus.getBlockSize());
            System.out.println(fileStatus.getPermission());//rw-r--r--
            System.out.println(fileStatus.getLen());
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                System.out.println("block-length:"+ blockLocation.getLength()+"--block-offset:"+ blockLocation.getOffset());
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println(host);
                }
            }
            System.out.println("----------------------------------------------");
        }
    }

    @Test
    public void testListAll()throws Exception{
        FileStatus[] fileStatuses = fs.listStatus(new Path("/wordcount"));
        for (FileStatus fileStatuse : fileStatuses) {
            if(fileStatuse.isFile()){
                System.out.println("file----------"+fileStatuse.getPath().getName());
            }else {
                System.out.println("dictionary------------"+fileStatuse.getPath().getName());
            }
        }
    }


    /**
     * 相对那些封装好的方法而言的更底层一些的操作方式
     * 上层的mr  spark等运算框架，去hdfs中获取数据的时候，就是调的底层的api
     *
     */
    @Test
    public void testDownLoadFileToLocal()throws Exception{
        //先获取一个文件的输入流---针对hdfs上的文件
        FSDataInputStream in = fs.open(new Path("/wordcount/hdfsTest.txt"));
        //再构造一个文件的输出流---针对本地的文件
        FileOutputStream out = new FileOutputStream(new File("d:/hdfs.txt"));
        //在将输入流中的数据传输到输出流
        IOUtils.copyBytes(in,out,4096);
    }

    /**
     * dhfs支持随机丁文进行文件读取，而且可以方便的读取指定长度
     * 用于上层分布式运算框架并发处理数据
     * @throws Exception
     */
    @Test
    public void testRandomAccess()throws Exception{
        FSDataInputStream in = fs.open(new Path("/wordcount/hdfsTest.txt"));
        //可以将留的其实偏移量进行自定义
        in.seek(22);
        FileOutputStream out = new FileOutputStream(new File("d:/hdfsseek.txt"));
        IOUtils.copyBytes(in,out,4096);
    }

    /**
     * 显示hdfs上文件的内容
     * @throws Exception
     */
    @Test
    public void testCat()throws Exception{
        Path path = new Path("/wordcount/hdfsTest.txt");
        FSDataInputStream in = fs.open(path);
        System.out.println("----------------------");
        IOUtils.copyBytes(in,System.out,1024);
        System.out.println("\t-------------------");

        FileStatus[] fileStatuses = fs.listStatus(path);
        BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatuses[0], 0L, fileStatuses[0].getLen());
        long length = fileBlockLocations[0].getLength();
        long offset = fileBlockLocations[0].getOffset();
        System.out.println("length:"+length+"---------offset:"+offset);
        IOUtils.copyBytes(in,System.out,(int)length);
        byte[] b = new byte[4096];
        FileOutputStream out = new FileOutputStream(new File("d:/block0"));
        while(in.read(offset,b,0,4096)!=-1){
            out.write(b);
            offset+=4096;
            if(offset>=length)
                return;
        }
        out.flush();
        out.close();
        in.close();

    }


}
